/*
 * Copyright 2002-2019 the original author or authors.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *      https://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package com.hsc.lock.interceptor;

import com.hsc.lock.interceptor.key.KeyGenerator;
import com.hsc.lock.interceptor.key.SimpleKeyGenerator;
import com.hsc.lock.interceptor.refuse.RefuseHandler;
import com.hsc.lock.interceptor.refuse.SimpleRefuseHandler;
import com.hsc.lock.interceptor.store.SimpleLockErrorHandler;
import com.hsc.lock.interceptor.store.StoreErrorHandler;
import com.hsc.lock.store.LockStore;
import com.hsc.lock.store.concurrent.ConcurrentMapCache;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.aop.framework.AopProxyUtils;
import org.springframework.aop.support.AopUtils;
import org.springframework.beans.factory.*;
import org.springframework.beans.factory.annotation.BeanFactoryAnnotationUtils;
import org.springframework.context.expression.AnnotatedElementKey;
import org.springframework.core.BridgeMethodResolver;
import org.springframework.expression.EvaluationContext;
import org.springframework.lang.Nullable;
import org.springframework.util.*;
import org.springframework.util.function.SingletonSupplier;
import org.springframework.util.function.SupplierUtils;

import java.lang.reflect.Method;
import java.lang.reflect.Proxy;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.function.Supplier;


public abstract class LockAspectSupport extends AbstractStoreInvoker
        implements BeanFactoryAware, InitializingBean, SmartInitializingSingleton {

    protected final Log logger = LogFactory.getLog(getClass());

    private final Map<OperationStoreKey, OperationMetadata> metadataLock = new ConcurrentHashMap<>(1024);

    private final OperationExpressionEvaluator evaluator = new OperationExpressionEvaluator();

    @Nullable
    private OperationSource operationSource;

    private SingletonSupplier<KeyGenerator> keyGenerator = SingletonSupplier.of(SimpleKeyGenerator::new);

    @Nullable
    private SingletonSupplier<RefuseHandler> refuseHandler;

    @Nullable
    private SingletonSupplier<LockStore> lockStore;

    @Nullable
    private BeanFactory beanFactory;

    private boolean initialized = false;

    public void configure(
            @Nullable Supplier<StoreErrorHandler> errorHandler, @Nullable Supplier<KeyGenerator> keyGenerator,
            @Nullable Supplier<RefuseHandler> refuseEnterHandler, @Nullable Supplier<LockStore> lockStore) {

        this.errorHandler = new SingletonSupplier<>(errorHandler, SimpleLockErrorHandler::new);
        this.keyGenerator = new SingletonSupplier<>(keyGenerator, SimpleKeyGenerator::new);
        this.refuseHandler = new SingletonSupplier<>(refuseEnterHandler, SimpleRefuseHandler::new);
        this.lockStore = new SingletonSupplier<>(lockStore, () ->
                new ConcurrentMapCache("ddd")
        );
    }


    public void setLockOperationSources(OperationSource... lockOperationSources) {
        Assert.notEmpty(lockOperationSources, "At least 1 LockOperationSource needs to be specified");
        this.operationSource = (lockOperationSources.length > 1 ?
                new CompositeLockOperationSource(lockOperationSources) : lockOperationSources[0]);
    }


    public void setLockOperationSource(@Nullable OperationSource lockOperationSource) {
        this.operationSource = lockOperationSource;
    }


    @Nullable
    public OperationSource getLockOperationSource() {
        return this.operationSource;
    }


    public void setKeyGenerator(KeyGenerator keyGenerator) {
        this.keyGenerator = SingletonSupplier.of(keyGenerator);
    }


    public KeyGenerator getKeyGenerator() {
        return this.keyGenerator.obtain();
    }


    @Nullable
    public LockStore getLockStore() {
        return SupplierUtils.resolve(this.lockStore);
    }

    @Nullable
    public SingletonSupplier<RefuseHandler> getRefuseEnterHandler() {
        return refuseHandler;
    }

    public void setRefuseEnterHandler(@Nullable SingletonSupplier<RefuseHandler> refuseEnterHandler) {
        this.refuseHandler = refuseEnterHandler;
    }

    @Override
    public void setBeanFactory(BeanFactory beanFactory) {
        this.beanFactory = beanFactory;
    }


    @Override
    public void afterPropertiesSet() {
        Assert.state(getLockOperationSource() != null, "The 'lockOperationSources' property is required: " +
                "If there are no cacheable methods, then don't use a cache aspect.");
    }

    @Override
    public void afterSingletonsInstantiated() {
        this.initialized = true;
    }


    protected String methodIdentification(Method method, Class<?> targetClass) {
        Method specificMethod = ClassUtils.getMostSpecificMethod(method, targetClass);
        return ClassUtils.getQualifiedMethodName(specificMethod);
    }


    protected OperationContext getOperationContext(
            ReplayLockOperation operation, Method method, Object[] args, Object target, Class<?> targetClass) {

        OperationMetadata metadata = getLockOperationMetadata(operation, method, targetClass);
        return new OperationContext(metadata, args, target);
    }


    protected OperationMetadata getLockOperationMetadata(
            ReplayLockOperation operation, Method method, Class<?> targetClass) {

        OperationStoreKey cacheKey = new OperationStoreKey(operation, method, targetClass);
        OperationMetadata metadata = this.metadataLock.get(cacheKey);
        if (metadata == null) {
            KeyGenerator operationKeyGenerator;
            if (StringUtils.hasText(operation.getKeyGenerator())) {
                operationKeyGenerator = getBean(operation.getKeyGenerator(), KeyGenerator.class);
            } else {
                operationKeyGenerator = getKeyGenerator();
            }
            LockStore lockStore = getLockStore();
            Assert.state(lockStore != null, "No LockResolver/LockManager set");

            metadata = new OperationMetadata(operation, method, targetClass,
                    operationKeyGenerator, lockStore);
            this.metadataLock.put(cacheKey, metadata);
        }
        return metadata;
    }


    protected <T> T getBean(String beanName, Class<T> expectedType) {
        if (this.beanFactory == null) {
            throw new IllegalStateException(
                    "BeanFactory must be set on cache aspect for " + expectedType.getSimpleName() + " retrieval");
        }
        return BeanFactoryAnnotationUtils.qualifiedBeanOfType(this.beanFactory, expectedType, beanName);
    }


    protected void clearMetadataLock() {
        this.metadataLock.clear();
        this.evaluator.clear();
    }

    @Nullable
    protected Object execute(OperationInvoker invoker, Object target, Method method, Object[] args) {
        // Check whether aspect is enabled (to cope with cases where the AJ is pulled in automatically)
        if (this.initialized) {
            Class<?> targetClass = getTargetClass(target);
            OperationSource lockOperationSource = getLockOperationSource();
            if (lockOperationSource != null) {
                Collection<ReplayLockOperation> operations = lockOperationSource.getCacheOperations(method, targetClass);
                if (!CollectionUtils.isEmpty(operations)) {
                    return execute(invoker, method,
                            new OperationContexts(operations, method, args, target, targetClass));
                }
            }
        }

        return invoker.invoke();
    }


    protected Object invokeOperation(OperationInvoker invoker) {
        return invoker.invoke();
    }

    private Class<?> getTargetClass(Object target) {
        return AopProxyUtils.ultimateTargetClass(target);
    }

    @Nullable
    private Object execute(final OperationInvoker invoker, Method method, OperationContexts contexts) {

        LockStore.ValueWrapper cacheHit = putIfAbsent(contexts.get(ReplayLockOperation.class), "1");
        Object returnValue;
        if (cacheHit == null) {
            try {
                returnValue = invokeOperation(invoker);
            } finally {
                processLockEvicts(contexts.get(ReplayLockOperation.class));
            }
        } else {
            throw new RuntimeException("");
        }
        return returnValue;
    }

    @Nullable
    private Object wrapLockValue(Method method, @Nullable Object cacheValue) {
        if (method.getReturnType() == Optional.class &&
                (cacheValue == null || cacheValue.getClass() != Optional.class)) {
            return Optional.ofNullable(cacheValue);
        }
        return cacheValue;
    }

    @Nullable
    private Object unwrapReturnValue(Object returnValue) {
        return ObjectUtils.unwrapOptional(returnValue);
    }


    private void performCacheEvict(OperationContext context, Operation operation, @Nullable Object result) {

//        Object key = null;
//        for (Cache cache : context.getCaches()) {
//            if (operation.isCacheWide()) {
//                logInvalidating(context, operation, null);
//                doClear(cache, operation.isBeforeInvocation());
//            }
//            else {
//                if (key == null) {
//                    key = generateKey(context, result);
//                }
//                logInvalidating(context, operation, key);
//                doEvict(cache, key, operation.isBeforeInvocation());
//            }
//        }
    }

    private void processLockEvicts(Collection<OperationContext> contexts) {

        for (OperationContext context : contexts) {
            OperationContext operation = context.metadata.operation;
            if (beforeInvocation == operation.isBeforeInvocation() && isConditionPassing(context, result)) {
                performLockEvict(context, operation, result);
            }
        }
    }


    /**
     * Find a cached item only for {@link com.hsc.lock.interceptor.LockableOperation} that passes the condition.
     *
     * @param contexts the cacheable operations
     * @return a {@link LockStore.ValueWrapper} holding the cached item,
     * or {@code null} if none is found
     */
    @Nullable
    private LockStore.ValueWrapper findLockdItem(Collection<OperationContext> contexts) {
        for (OperationContext context : contexts) {
            Object key = generateKey(context);
            LockStore.ValueWrapper cached = findInLocks(context, key);
            if (cached != null) {
                return cached;
            } else {
                if (logger.isTraceEnabled()) {
                    logger.trace("No cache entry for key '" + key + "' in cache(s) ");
                }
            }

        }
        return null;
    }


    @Nullable
    private LockStore.ValueWrapper putIfAbsent(Collection<OperationContext> contexts, Object value) {
        for (OperationContext context : contexts) {
            Object key = generateKey(context);
            LockStore.ValueWrapper cached = putIfAbsent(context, key, value);
            if (cached != null) {
                return cached;
            } else {
                if (logger.isTraceEnabled()) {
                    logger.trace("No cache entry for key '" + key + "' in cache(s) ");
                }
            }

        }
        return null;
    }


    @Nullable
    private LockStore.ValueWrapper findInLocks(OperationContext context, Object key) {
        for (LockStore cache : Arrays.asList(context.getLocks())) {
            LockStore.ValueWrapper wrapper = doGet(cache, key);
            if (wrapper != null) {
                if (logger.isTraceEnabled()) {
                    logger.trace("Lock entry for key '" + key + "' found in cache '" + "'");
                }
                return wrapper;
            }
        }
        return null;
    }

    @Nullable
    private LockStore.ValueWrapper putIfAbsent(OperationContext context, Object key, Object value) {
        for (LockStore cache : Arrays.asList(context.getLocks())) {
            LockStore.ValueWrapper wrapper = doPutIfAbsent(cache, key, value);
            if (wrapper != null) {
                if (logger.isTraceEnabled()) {
                    logger.trace("Lock entry for key '" + key + "' found in cache '" + "'");
                }
                return wrapper;
            }
        }
        return null;
    }


    private Object generateKey(OperationContext context) {
        Object key = context.generateKey();
        if (key == null) {
            throw new IllegalArgumentException("Null key returned for cache operation (maybe you are " +
                    "using named params on classes without debug info?) " + context.metadata.operation);
        }
        if (logger.isTraceEnabled()) {
            logger.trace("Computed cache key '" + key + "' for operation " + context.metadata.operation);
        }
        return key;
    }


    private class OperationContexts {

        private final MultiValueMap<Class<? extends Operation>, OperationContext> contexts;

        private final boolean sync;

        public OperationContexts(Collection<? extends ReplayLockOperation> operations, Method method,
                                     Object[] args, Object target, Class<?> targetClass) {

            this.contexts = new LinkedMultiValueMap<>(operations.size());
            for (ReplayLockOperation op : operations) {
                this.contexts.add(op.getClass(), getOperationContext(op, method, args, target, targetClass));
            }
            this.sync = determineSyncFlag(method);
        }

        public Collection<OperationContext> get(Class<? extends ReplayLockOperation> operationClass) {
            Collection<OperationContext> result = this.contexts.get(operationClass);
            return (result != null ? result : Collections.emptyList());
        }

        public boolean isSynchronized() {
            return this.sync;
        }

        private boolean determineSyncFlag(Method method) {
            return false;
        }
    }


    protected static class OperationMetadata {

        private final Operation operation;

        private final Method method;

        private final Class<?> targetClass;

        private final Method targetMethod;

        private final AnnotatedElementKey methodKey;

        private final KeyGenerator keyGenerator;

        private final LockStore lockStore;


        public OperationMetadata(Operation operation, Method method, Class<?> targetClass,
                                 KeyGenerator keyGenerator, LockStore lockStore) {

            this.operation = operation;
            this.method = BridgeMethodResolver.findBridgedMethod(method);
            this.targetClass = targetClass;
            this.targetMethod = (!Proxy.isProxyClass(targetClass) ?
                    AopUtils.getMostSpecificMethod(method, targetClass) : this.method);
            this.methodKey = new AnnotatedElementKey(this.targetMethod, targetClass);
            this.keyGenerator = keyGenerator;
            this.lockStore = lockStore;
        }
    }


    protected class OperationContext implements OperationInvocationContext<Operation> {

        private final OperationMetadata metadata;

        private final Object[] args;

        private final Object target;


        public OperationContext(OperationMetadata metadata, Object[] args, Object target) {
            this.metadata = metadata;
            this.args = extractArgs(metadata.method, args);
            this.target = target;
        }

        @Override
        public Operation getOperation() {
            return this.metadata.operation;
        }

        @Override
        public Object getTarget() {
            return this.target;
        }

        @Override
        public Method getMethod() {
            return this.metadata.method;
        }

        @Override
        public Object[] getArgs() {
            return this.args;
        }

        private Object[] extractArgs(Method method, Object[] args) {
            if (!method.isVarArgs()) {
                return args;
            }
            Object[] varArgs = ObjectUtils.toObjectArray(args[args.length - 1]);
            Object[] combinedArgs = new Object[args.length - 1 + varArgs.length];
            System.arraycopy(args, 0, combinedArgs, 0, args.length - 1);
            System.arraycopy(varArgs, 0, combinedArgs, args.length - 1, varArgs.length);
            return combinedArgs;
        }


        @Nullable
        protected Object generateKey() {
            if (StringUtils.hasText(this.metadata.operation.getKey())) {
                EvaluationContext evaluationContext = createEvaluationContext();
                return evaluator.key(this.metadata.operation.getKey(), this.metadata.methodKey, evaluationContext);
            }
            return this.metadata.keyGenerator.generate(this.target, this.metadata.method, this.args);
        }

        private EvaluationContext createEvaluationContext() {
            return evaluator.createEvaluationContext(this.metadata.lockStore, this.metadata.method, this.args,
                    this.target, this.metadata.targetClass, this.metadata.targetMethod, beanFactory);
        }

        protected LockStore getLocks() {
            return this.metadata.lockStore;
        }

    }


    private static final class OperationStoreKey {

        private final Operation operation;

        private final AnnotatedElementKey methodLockKey;

        private OperationStoreKey(Operation operation, Method method, Class<?> targetClass) {
            this.operation = operation;
            this.methodLockKey = new AnnotatedElementKey(method, targetClass);
        }

        @Override
        public boolean equals(@Nullable Object other) {
            if (this == other) {
                return true;
            }
            if (!(other instanceof OperationStoreKey)) {
                return false;
            }
            OperationStoreKey otherKey = (OperationStoreKey) other;
            return (this.operation.equals(otherKey.operation) &&
                    this.methodLockKey.equals(otherKey.methodLockKey));
        }

        @Override
        public int hashCode() {
            return (this.operation.hashCode() * 31 + this.methodLockKey.hashCode());
        }

        @Override
        public String toString() {
            return this.operation + " on " + this.methodLockKey;
        }
    }

}
